#ifndef UL_FIFO_TASK_QUEUE_H_ 
#define UL_FIFO_TASK_QUEUE_H_
#include <pthread.h>
#include "ul_define.h"
#include "ul_malloc.h"
namespace LinuxNetworkUtility
{
  /// one producer, multi consumer, thread safe and lock free
  class UlFIFO
  {
  public:
    UlFIFO();
    ~UlFIFO();
    int init(const int64_t queue_size, const int64_t consumer_num); 
    typedef enum
    {
      FETCH_EVERY_MSG = 1,
      FETCH_UNFETCHED_MSG = 2,
      FETCH_BLOCKING = 4
    }FetchFlag;
    /// this function will block until there is msg if FETCH_BLOCKING was set
    void *pop(const int64_t consumer_id, const FetchFlag flag);
    /// this function will return immediately if queue is full
    int push(void *msg);
  private:
    void destroy();
    class FIFOMsgHead
    {
    public:
      FIFOMsgHead()
      {
        unfected_consumer_num_ = 0;
        msg_ = NULL;
        fetched_ = 0;
      }
      ~FIFOMsgHead()
      {
        unfected_consumer_num_ = 0;
        msg_ = NULL;
        fetched_ = 0;
      }
      inline void init(const int64_t consumer_num, void *msg)
      {
        unfected_consumer_num_ = consumer_num;
        msg_ = msg;
        fetched_ = 0;
      }
      /// fetch msg
      void *fetch_msg(const FetchFlag flag, int32_t & unfetched_consumer_num);

    private:
      volatile int32_t  fetched_;
      volatile int32_t  unfected_consumer_num_;
      void *            msg_;
    };

    struct FIFOConsumerInfo
    {
      FIFOConsumerInfo()
      {
        pop_count_ = 0;
        pop_idx_ = 0;
      }
      uint64_t pop_count_;
      int64_t pop_idx_;
    };

    FIFOMsgHead         *msg_queue_;
    int64_t             queue_size_;
    int64_t             max_idx_;

    uint64_t            push_count_;
    int64_t             push_idx_;
    uint64_t            pop_count_;
    int64_t             pop_idx_;

    int64_t             consumer_num_;
    FIFOConsumerInfo    *consumers_;

    int32_t             has_waiting_consumer_;
    pthread_mutex_t     cond_mutex_;
    pthread_cond_t      cond_;
  };

  class UlFIFOQueenWorkerTQ
  {
  public:
    UlFIFOQueenWorkerTQ();
    virtual ~UlFIFOQueenWorkerTQ();

    virtual int init(const int64_t worker_bee_num, const int64_t queue_size);
    /// the queen bee dispatch task to worker bees
    virtual int dispatch_task(void *tsk);
    /// the worker bee claim task to do
    virtual int claim_task(void **tsk, const int64_t ntsk, const int64_t worker_id, 
                           const UlFIFO::FetchFlag flag);
    /// the worker bee submit output of task
    virtual int submit_output(void *tsk, const int64_t worker_id);
    /// the queen bee fetch output of tasks
    virtual int fetch_output(void **tsk, const int64_t ntsk);

    virtual void* malloc_tsk(const int64_t size, const int32_t mod_id)
    {
      return ul_malloc(size,mod_id,NULL, FREE_TO_ALLOCATED_THREAD_POOL);
    } 
    virtual void free_tsk(void* tsk)
    {
      ul_free(tsk);
    }
  private:
    void destroy();
    int64_t worker_num_;
    int64_t prev_fetch_worker_idx_;
    UlFIFO  queen_end_fifo_;
    UlFIFO  *worker_end_fifos_;
  };
}   
#endif /* UL_FIFO_TASK_QUEUE_H_ */
